package com.relytosoft.mizar.analyze.system.services.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.relytosoft.dubhe.common.kafka.producer.KafkaOpMsg;
import com.relytosoft.mizar.analyze.base.constant.MediaMqConstants;
import com.relytosoft.mizar.analyze.base.entity.pojo.dto.FacePersonResDto;
import com.relytosoft.mizar.analyze.system.api.entity.AyPersonGroup;
import com.relytosoft.mizar.analyze.system.services.service.AyGroupBaseService;
import com.relytosoft.mizar.analyze.system.services.service.AyPersonGroupService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author lj
 * @date 2022/5/18 9:07
 */
@Component
public class AyPersonInfoConsumer {
    protected Logger log = LoggerFactory.getLogger(this.getClass());

    @Resource
    private AyGroupBaseService ayGroupBaseService;

    @Resource
    private AyPersonGroupService ayPersonGroupService;

    @KafkaListener(id = "consumer-mizar-person-sync", topics = MediaMqConstants.FACE_ANALYZE_PERSON_ID_TOPIC, containerFactory =
            "singleContainerFactory")
    public void processSingleMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        try {
            KafkaOpMsg kafkaOpMsg = JSON.parseObject(record.value().toString(), KafkaOpMsg.class);
            boolean isConsume = false;
            String data = kafkaOpMsg.getData();
            Map<String, Object> map = JSONObject.parseObject(data, new TypeReference<Map<String, Object>>() {
            });
            //自己系统人像ID
            Integer faceId = (Integer) map.get("faceId");
            //btdl返回人脸和库数据
            Object personInfos = map.get("personInfos");
            if (personInfos != null) {
                List<FacePersonResDto> dtoList = JSONObject.parseObject(JSONObject.toJSONString(personInfos), new TypeReference<List<FacePersonResDto>>() {
                });
                List<String> groupIds = dtoList.stream().map(FacePersonResDto::getGroupId).collect(Collectors.toList());
                Map<String, Integer> groupIdMap = ayGroupBaseService.selectPlateformAndGroupId(groupIds);
                Set<Integer> ownGroupIds = new HashSet<>(groupIdMap.values());
                //查询faceId和ownGroupIds的数据
                List<AyPersonGroup> ayPersonGroupList = ayPersonGroupService.selectAyPersonGroup(ownGroupIds, faceId);
                for (FacePersonResDto facePersonResDto : dtoList) {
                    for (AyPersonGroup ayPersonGroup : ayPersonGroupList) {
                        if (groupIdMap.get(facePersonResDto.getGroupId()).equals(ayPersonGroup.getGroupId())) {
                            ayPersonGroup.setPlatformPersonId(facePersonResDto.getPersonId());
                        }
                    }
                }
                isConsume = ayPersonGroupService.updateBatchById(ayPersonGroupList);
            }
            log.info("线程:{} 消费->{}", Thread.currentThread(), kafkaOpMsg.getOpType());
            if (isConsume) {
                ack.acknowledge();
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error(e.getMessage());
        }
    }
}
